package com.starnet.yarnmonitor.yarn.schedule;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.starnet.yarnmonitor.yarn.dto.KafkaType;
import com.starnet.yarnmonitor.yarn.dto.KafkaWarn;
import com.starnet.yarnmonitor.yarn.entity.YarnWarnHistory;
import com.starnet.yarnmonitor.yarn.entity.YarnWarnOption;
import com.starnet.yarnmonitor.yarn.mapper.YarnWarnHistoryMapper;
import com.starnet.yarnmonitor.yarn.mapper.YarnWarnOptionMapper;
import com.starnet.yarnmonitor.yarn.service.ClusterService;
import com.starnet.yarnmonitor.yarn.utils.KafkaConsumerUtil;
import com.starnet.yarnmonitor.yarn.utils.KafkaProducerUtil;
import com.starnet.yarnmonitor.yarn.vo.AppVO;
import com.starnet.yarnmonitor.yarn.vo.MetricsVO;
import com.starnet.yarnmonitor.yarn.vo.NodeVO;
import com.starnet.yarnmonitor.yarn.vo.QueueVO;
import lombok.extern.log4j.Log4j;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;

/***
 * 要加一个新的指标 先在manger（）内注册自动恢复，在对应的类型里加监控代码
 */
@Log4j
@Component
public class MonitorSchedule {
    @Autowired
    ClusterService clusterService;
    @Autowired
    YarnWarnOptionMapper yarnWarnOptionMapper;
    @Autowired
    YarnWarnHistoryMapper yarnWarnHistoryMapper;
    @Autowired
    KafkaConsumerUtil kafkaConsumerUtil;
    @Autowired
    KafkaProducerUtil kafkaProducerUtil;

    private static HashMap<String, Integer> autoRecoverMap = new HashMap<>();

    /***
     * 每次新增一个监控指标就要来这边注册一个
     */
    @PostConstruct
    public void manager() {
        List<NodeVO> allNodes = null;
        List<QueueVO> allQueues=null;
        try {
            allNodes = clusterService.getAllNodes(null, null);
            allQueues = clusterService.getAllQueues(null, null);
        } catch (Exception e){
            e.printStackTrace();
            log.error(e.getStackTrace());
        }

        List<AppVO> allApps = null;
        try {
            allApps = clusterService.getAllApps(null, null);
        } catch (Exception e){
            e.printStackTrace();
            log.error(e.getStackTrace());
        }
        register("clusterAppFail");
        register("clusterUsedMemory");
        register("clusterVCores");
        register("clusterUnhealthyNodes");
        register("clusterShutdownNodes");
        //节点部分
        for (NodeVO nodeVO : allNodes) {
            if (allNodes==null){
                break;
            }
            String nodeId = nodeVO.getId();
            //注册节点内存自动恢复
            String nodeUsedMemoryRecoveryKey = getRecoveryKey(6, nodeId);
            register(nodeUsedMemoryRecoveryKey);
            //注册节点虚拟核心自动恢复
            String nodeUsedVCores = getRecoveryKey(7, nodeId);
            register(nodeUsedVCores);
        }
        for (QueueVO queueVO : allQueues) {
            if (allQueues==null){
                break;
            }
            String queueName = queueVO.getQueueName();
            //监控队列容量
            String nodeCapacityRecoveryKey = getRecoveryKey(8, queueName);
            register(nodeCapacityRecoveryKey);
            //监控队列内存使用率
            String queueMemoryRecoveryKey = getRecoveryKey(9, queueName);
            register(queueMemoryRecoveryKey);
            //监控队列虚拟核心使用率
            String queueVCoresRecoveryKey = getRecoveryKey(10, queueName);
            register(queueVCoresRecoveryKey);
        }

    }


    /***\
     * 监控集群
     */
    @Scheduled(cron = "0/2 * * * * ? ")
    public void clusterMonitor() {
        //监控集群App失败
        long clusterAppFailOptionId = 1L;
        MetricsVO clusterInfo = null;
        try {
            clusterInfo = clusterService.getClusterInfo();
        } catch (IOException e) {
            log.error(e.getStackTrace());
            e.printStackTrace();
        }
        double appsFailed = clusterInfo.getAppsFailed();
        gtWarn(clusterAppFailOptionId, appsFailed, 3, "clusterAppFail", "集群");
        //监控集群已用内存
        long clusterUsedMemoryOptionId = 2L;
        Integer allocatedMB = clusterInfo.getAllocatedMB();
        Integer availableMB = clusterInfo.getAvailableMB();
        if (availableMB != 0) {
            double clusterUsedMemoryRatio = (double) allocatedMB / (double) availableMB;
            //保留两位小数
            clusterUsedMemoryRatio = new BigDecimal(clusterUsedMemoryRatio).setScale(2, BigDecimal.ROUND_DOWN).doubleValue();

            gtWarn(clusterUsedMemoryOptionId, clusterUsedMemoryRatio, 3, "clusterUsedMemory", "集群");

        }
        //监控集群已用虚拟核心数
        long clusterUsedVCoresOptionId = 3L;
        Integer allocatedVirtualCores = clusterInfo.getAllocatedVirtualCores();
        Integer availableVirtualCores = clusterInfo.getAvailableVirtualCores();
        if (availableVirtualCores != 0) {
            double clusterUsedVCoresRatio = (double) allocatedVirtualCores / (double) availableVirtualCores;
            clusterUsedVCoresRatio = new BigDecimal(clusterUsedVCoresRatio).setScale(2, BigDecimal.ROUND_DOWN).doubleValue();
            gtWarn(clusterUsedVCoresOptionId, clusterUsedVCoresRatio, 3, "clusterVCores", "集群");
        }

        //监控集群不健康节点数量
        //monitor表id
        long clusterUnhealthyNodesOptionId = 4L;
        Integer unhealthyNodes = clusterInfo.getUnhealthyNodes();
        //public void gtWarn(long optionId, double monitorValue, int recoveryThreshold, Integer recoveryCur)
        gtWarn(clusterUnhealthyNodesOptionId, unhealthyNodes, 3, "clusterUnhealthyNodes", "集群");


        //监控集群终止节点
        long clusterShutdownNodesOptionId = 5L;
        Integer shutdownNodes = clusterInfo.getShutdownNodes();
        //public void gtWarn(long optionId, double monitorValue, int recoveryThreshold, Integer recoveryCur)
        gtWarn(clusterShutdownNodesOptionId, shutdownNodes, 3, "clusterShutdownNodes", "集群");
    }

    /***
     * 监控集群的虚拟核心
     */
    @Scheduled(cron = "0/2 * * * * ? ")
    public void nodeMonitor() {
        //节点已使用内存阈值id
        long nodeUsedMemoryOptionId = 6L;
        //节点已使用核心阈值id
        long nodeUsedVirtualCoresOptionId = 7L;
        //获取所有节点
        List<NodeVO> allNodes = null;
        try {
            allNodes = clusterService.getAllNodes(null, null);
        } catch (IOException e) {
            e.printStackTrace();
            log.error(e.getStackTrace());
        }
        for (NodeVO nodeVO : allNodes) {
            if (allNodes==null){
                break;
            }
            String nodeId = nodeVO.getId();
            //监控节点已使用内存
            Integer availMemoryMB = nodeVO.getAvailMemoryMB();
            Integer usedMemoryMB = nodeVO.getUsedMemoryMB();
            if (availMemoryMB != 0) {
                double usedMemoryMBRatio = (double) usedMemoryMB / (double) availMemoryMB;
                usedMemoryMBRatio = new BigDecimal(usedMemoryMBRatio).setScale(2, BigDecimal.ROUND_DOWN).doubleValue();
                String nodeUsedMemoryRecoveryKey = getRecoveryKey(nodeUsedMemoryOptionId, nodeId);
                gtWarn(nodeUsedMemoryOptionId, usedMemoryMBRatio, 3, nodeUsedMemoryRecoveryKey, nodeId);
            }
            //监控节点虚拟核心
            Integer usedVirtualCores = nodeVO.getUsedVirtualCores();
            Integer availableVirtualCores = nodeVO.getAvailableVirtualCores();
            if (availableVirtualCores != 0) {
                double usedVirtualCoresRatio = (double) usedVirtualCores / (double) availableVirtualCores;
                String clusterUsedMemoryRecoveryKey = getRecoveryKey(nodeUsedVirtualCoresOptionId, nodeId);
                gtWarn(nodeUsedVirtualCoresOptionId, usedVirtualCoresRatio, 3, clusterUsedMemoryRecoveryKey, nodeId);

            }
        }


    }

    /***
     * 监控队列
     */
    @Scheduled(cron = "0/2 * * * * ? ")
    public void queueMonitor() {
        //队列容量阈值id
        long queueCapacityOptionId = 8L;
        //队列内存使用率阈值id
        long queueMemoryOptionId = 9L;
        //队列虚拟核心使用率阈值id
        long queueVCoresOptionId = 9L;
        //获取所有队列
        List<QueueVO> allQueues = null;
        try {
            allQueues = clusterService.getAllQueues(null, null);
        } catch (Exception e){
            e.printStackTrace();
            log.info(e.getStackTrace());
        }
        for (QueueVO queue : allQueues) {
            if (allQueues==null){
                break;
            }
            //队列唯一标识
            String warnId = queue.getQueueName();
            //监控队列容量
            double capacity = queue.getCapacity();
            double maximumCapacity = queue.getMaximumCapacity();
            if (maximumCapacity != 0) {
                double capacityRatio = capacity / maximumCapacity;
                System.out.println(maximumCapacity);
                System.out.println(capacityRatio);
                capacityRatio = new BigDecimal(capacityRatio).setScale(2, BigDecimal.ROUND_DOWN).doubleValue();
                String queueCapacityRecoveryKey = getRecoveryKey(queueCapacityOptionId, warnId);
                gtWarn(queueCapacityOptionId, capacityRatio, 3, queueCapacityRecoveryKey, warnId);
            }

            //监控队列内存使用率
            Long allocatedMemoryMB = queue.getAllocatedMemoryMB();
            Long availableMemoryMB = queue.getAvailableMemoryMB();
            if (availableMemoryMB != 0) {
                double queueMemoryRatio = (double) allocatedMemoryMB / (double) availableMemoryMB;
                queueMemoryRatio = new BigDecimal(queueMemoryRatio).setScale(2, BigDecimal.ROUND_DOWN).doubleValue();
                String queueMemoryRecoveryKey = getRecoveryKey(queueMemoryOptionId, warnId);
                gtWarn(queueMemoryOptionId, queueMemoryRatio, 3, queueMemoryRecoveryKey, warnId);
            }

            //监控队列虚拟核心使用率
            Long allocatedVCores = queue.getAllocatedVCores();
            Long availableVCores = queue.getAvailableVCores();
            if (availableVCores != 0) {
                double queueVCoresRatio = (double) allocatedVCores / availableVCores;
                String queueVCoresRecoveryKey = getRecoveryKey(queueVCoresOptionId, warnId);
                gtWarn(queueVCoresOptionId, queueVCoresRatio, 3, queueVCoresRecoveryKey, warnId);
            }


        }

    }

    /***
     * 监控应用
     */
    @Scheduled(cron = "0/2 * * * * ? ")
    public void appMonitor() {
        long appFailOptionId = 11L;
        List<AppVO> allApps = null;
        try {
            allApps = clusterService.getAllApps(null, null);
        } catch (IOException e) {
            log.error(e.getMessage());
            e.printStackTrace();
        } catch (YarnException e) {
            log.error(e.getMessage());
            e.printStackTrace();
        }
        if (allApps!=null){
            for (AppVO appVo : allApps) {
                appFailWarn(appFailOptionId,appVo);
            }
        }

    }

    /***
     * 注册恢复值
     * @param autoRecoverKey 通用方法内获取恢复值的key
     */
    public static void register(String autoRecoverKey) {
        autoRecoverMap.put(autoRecoverKey, 4);
    }

    /***
     * 获取自动恢复的key 通过拼接预警id和同一个预警中的唯一标识得到
     * @param optionId 预警表id
     * @param warnId 同一个预警设置的不同节点（队列、app）唯一标识
     * @return
     */
    public static String getRecoveryKey(long optionId, String warnId) {
        String recoveryKey = "Option:" + optionId + ";" + "WarnId:" + warnId;
        return recoveryKey;
    }

    /***
     * 监控app失败杀死
     */
    public void appFailWarn(long optionId, AppVO appVo) {
        YarnWarnOption yarnWarnOption = yarnWarnOptionMapper.selectById(optionId);
        if (yarnWarnOption == null) {
            log.error("告警阈值设置表中不存在id为" + optionId + "的值");
            return;
        }
        //warnId用app的id
        boolean isFail = appVo.getFinal_application_status().equalsIgnoreCase("killed") || appVo.getFinal_application_status().equalsIgnoreCase("undefined");
        if (!isFail) {
            return;
        }
        LambdaQueryWrapper<YarnWarnHistory> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(YarnWarnHistory::getWarnId, appVo.getApplicationId());
        queryWrapper.eq(YarnWarnHistory::getOptionId, optionId);
        YarnWarnHistory yarnWarnHistory = yarnWarnHistoryMapper.selectOne(queryWrapper);
        if (yarnWarnHistory == null) {
            yarnWarnHistory = new YarnWarnHistory();
            yarnWarnHistory.setOptionId(optionId);
            yarnWarnHistory.setWarnId(appVo.getApplicationId());
            yarnWarnHistory.setWarnType(yarnWarnOption.getWarnType());
            yarnWarnHistory.setNowValue(0.0);
            yarnWarnHistory.setLimitValue(0.0);
            yarnWarnHistory.setMsg("应用" + appVo.getApplicationId() + "执行失败，状态为：" + appVo.getFinal_application_status());
            yarnWarnHistory.setIsRecovered(false);
            yarnWarnHistory.setCreatedTime(Timestamp.valueOf(LocalDateTime.now()));
            yarnWarnHistory.setLastWarnTime(Timestamp.valueOf(LocalDateTime.now()));
            int insert = yarnWarnHistoryMapper.insert(yarnWarnHistory);
            if (insert != 1) {
                log.error("应用" + appVo.getApplicationId() + "运行失败但告警写入数据库失败");
                return;
            }
            log.warn("应用" + appVo.getApplicationId() + "运行失败");
        }
    }

    /***
     *
     * @param optionId option表里面对应的阈值的主键
     * @param monitorValue 当前监控到的值
     * @param recoveryThreshold 正常几次后自动恢复
     * @param autoRecoverKey 现在正常几次了
     */
    //大于等于的监控方法
    public void gtWarn(long optionId, double monitorValue, int recoveryThreshold, String autoRecoverKey, String warnID) {
        if (!autoRecoverMap.containsKey(autoRecoverKey)) {
            log.warn(Thread.currentThread().getStackTrace() + ":传入参数autoRecoverKey错误---optionId:" + optionId);
            return;
        }
        int recoveryCur = autoRecoverMap.get(autoRecoverKey);

        YarnWarnOption yarnWarnOption = yarnWarnOptionMapper.selectById(optionId);
        //如果这个阈值设置被人删了或者关了  就把这个方法通过返回中断
        if (yarnWarnOption == null || yarnWarnOption.getIsOpen() == false) {
            return;
        }
        //阈值
        double limit = yarnWarnOption.getValue();
        //如果大于等于阈值
        if (monitorValue >= limit) {
            //构建告警并发送
            YarnWarnHistory warn = new YarnWarnHistory();
            warn.setOptionId(optionId);
            warn.setWarnId(warnID);
            warn.setCreatedTime(Timestamp.valueOf(LocalDateTime.now()));
            warn.setLastWarnTime(Timestamp.valueOf(LocalDateTime.now()));
            warn.setIsRecovered(false);
            warn.setLimitValue(limit);
            warn.setWarnType(yarnWarnOption.getWarnType());
            warn.setMsg(warnID + yarnWarnOption.getDescription());
            warn.setNowValue(Double.valueOf(monitorValue));
            KafkaWarn kafkaWarn =new KafkaWarn(KafkaType.WARN,warn);
            kafkaProducerUtil.send("yarn-warn", JSON.toJSONString(kafkaWarn));
            //自动恢复计数清零
            autoRecoverMap.put(autoRecoverKey, 0);
        }
        //当前已恢复正常
        if (monitorValue < limit) {
            if (recoveryCur < recoveryThreshold) {
                //检测到一次 但没达到自动恢复的阈值 就计一次
                autoRecoverMap.put(autoRecoverKey, recoveryCur + 1);
                return;
            }
            if (recoveryCur == recoveryThreshold) {
                //达到恢复阈值，发送恢复消息
                autoRecoverMap.put(autoRecoverKey, recoveryCur + 1);
                YarnWarnHistory  yarnWarnHistory = new YarnWarnHistory();
                yarnWarnHistory.setWarnId(warnID);
                yarnWarnHistory.setOptionId(optionId);
                yarnWarnHistory.setRecoveryTime(Timestamp.valueOf(LocalDateTime.now()));
                yarnWarnHistory.setLimitValue(limit);
                yarnWarnHistory.setNowValue(Double.valueOf(monitorValue));
                KafkaWarn kafkaWarn =new KafkaWarn(KafkaType.RECOVER,yarnWarnHistory);
                kafkaProducerUtil.send("yarn-warn", JSON.toJSONString(kafkaWarn));
            }
        }

    }

    /***
     *
     * @param optionId option表里面对应的阈值的主键
     * @param monitorValue 当前监控到的值
     * @param recoveryThreshold 正常几次后自动恢复
     * @param autoRecoverKey 现在正常几次了
     */
    //小于等于的监控方法
    public void ltWarn(long optionId, double monitorValue, int recoveryThreshold, String autoRecoverKey, String warnID) {
        if (!autoRecoverMap.containsKey(autoRecoverKey)) {
            log.warn(Thread.currentThread().getStackTrace() + ":传入参数autoRecoverKey错误---optionId:" + optionId);
            return;
        }
        int recoveryCur = autoRecoverMap.get(autoRecoverKey);

        YarnWarnOption yarnWarnOption = yarnWarnOptionMapper.selectById(optionId);
        //如果这个阈值设置被人删了或者关了  就把这个方法通过返回中断
        if (yarnWarnOption == null || yarnWarnOption.getIsOpen() == false) {
            return;
        }
        //阈值
        double limit = yarnWarnOption.getValue();
        //如果大于等于阈值
        if (monitorValue <= limit) {
            //自动恢复计数清零
            autoRecoverMap.put(autoRecoverKey, 0);
            //查看数据库中有没有未恢复的 有就更新
            LambdaQueryWrapper<YarnWarnHistory> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(YarnWarnHistory::getWarnId, warnID);
            queryWrapper.eq(YarnWarnHistory::getOptionId, optionId);
            queryWrapper.eq(YarnWarnHistory::getIsRecovered, false);
            YarnWarnHistory yarnWarnHistory = yarnWarnHistoryMapper.selectOne(queryWrapper);
            //没有就新增
            if (yarnWarnHistory == null) {
                YarnWarnHistory warn = new YarnWarnHistory();
                warn.setOptionId(optionId);
                warn.setWarnId(warnID);
                warn.setCreatedTime(Timestamp.valueOf(LocalDateTime.now()));
                warn.setLastWarnTime(Timestamp.valueOf(LocalDateTime.now()));
                warn.setIsRecovered(false);
                warn.setLimitValue(limit);
                warn.setWarnType(yarnWarnOption.getWarnType());
                warn.setMsg(warnID + yarnWarnOption.getDescription());
                warn.setNowValue(Double.valueOf(monitorValue));
                int insert = yarnWarnHistoryMapper.insert(warn);
                log.info("新增告警"+warn);
                if (insert != 1) {
                    log.info("异常写入数据库失败");
                }
            } else {
                //如果有,就更新.更新当前值和心跳时间
                yarnWarnHistory.setLastWarnTime(Timestamp.valueOf(LocalDateTime.now()));
                yarnWarnHistory.setNowValue(monitorValue);
                yarnWarnHistory.setLimitValue(limit);
                yarnWarnHistoryMapper.updateById(yarnWarnHistory);
                log.info("更新告警"+yarnWarnHistory);
            }

        }
        //当前有app失败数量已恢复正常
        if (monitorValue > limit) {
            if (recoveryCur < recoveryThreshold) {
                //检测到一次 但没达到自动恢复的阈值 就计一次
                autoRecoverMap.put(autoRecoverKey, recoveryCur + 1);
                return;
            }
            if (recoveryCur >= recoveryThreshold) {
                //检测到一次正常 达到自动恢复阈值 就恢复 并且清零阈值
                autoRecoverMap.put(autoRecoverKey, 0);
            }
            //如果有未恢复的预警，恢复一下,如果没有未恢复的预警，就跳过这次
            LambdaQueryWrapper<YarnWarnHistory> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(YarnWarnHistory::getWarnId, warnID);
            queryWrapper.eq(YarnWarnHistory::getOptionId, optionId);
            queryWrapper.eq(YarnWarnHistory::getIsRecovered, false);
            YarnWarnHistory yarnWarnHistory = yarnWarnHistoryMapper.selectOne(queryWrapper);
            //没有未恢复的预警，跳过这次
            if (yarnWarnHistory == null) {
                return;
            }
            //有正在告警的，恢复
            yarnWarnHistory.setLimitValue(limit);
            yarnWarnHistory.setIsRecovered(true);
            yarnWarnHistory.setRecoveryTime(Timestamp.valueOf(LocalDateTime.now()));
            int i = yarnWarnHistoryMapper.updateById(yarnWarnHistory);
            if (i!=1){
                log.error("自动恢复告警失败,告警:"+yarnWarnHistory);
            }else {
                log.info("自动恢复告警成功，告警内容："+yarnWarnHistory);
            }

        }

    }

}
